package com.reactor.回压;

import java.util.concurrent.TimeUnit;

import org.junit.Test;
import org.reactivestreams.Subscription;

import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;

public class Demo {

	@Test
	// 假设，我们现在有一个非常快的Publisher——Flux.range(1, 6)，然后自定义一个每秒处理一个数据元素的慢的Subscriber，Subscriber就需要通过request(n)的方法来告知上游它的需求速度。代码如下：
	public void testBackpressure() {
	    Flux.range(1, 6)    // Flux.range是一个快的Publisher
	            .doOnRequest(n -> System.out.println("Request " + n + " values..."))    // 在每次request的时候打印request个数
	            .subscribe(new BaseSubscriber<Integer>() {  // 通过重写BaseSubscriber的方法来自定义Subscriber
	                @Override
	                protected void hookOnSubscribe(Subscription subscription) { // hookOnSubscribe定义在订阅的时候执行的操作
	                    System.out.println("Subscribed and make a request...");
	                    request(1); // 订阅时首先向上游请求1个元素
	                }

	                @Override
	                protected void hookOnNext(Integer value) {  // hookOnNext定义每次在收到一个元素的时候的操作
	                    try {
	                        TimeUnit.SECONDS.sleep(1);  // sleep 1秒钟来模拟慢的Subscriber
	                    } catch (InterruptedException e) {
	                        e.printStackTrace();
	                    }
	                    System.out.println("Get value [" + value + "]");    // 打印收到的元素
	                    request(1); // 每次处理完1个元素后再请求1个
	                }
	            });
	}
	// 这6个元素是以每秒1个的速度被处理的。由此可见range方法生成的Flux采用的是缓存的回压策略，能够缓存下游暂时来不及处理的元素。

}
